package com.atguigu.wc.source;

import com.atguigu.wc.beans.SensorReading;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.util.HashMap;
import java.util.Random;

/**
 * @Author:GaoFei
 * @Description:
 * @Date:Created in 18:11
 * @Modified By: 自定义实现数据源  实现SourceFunction接口即可
 */
public class SourceDefined {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<SensorReading> sensorReadingDataStreamSource = env.addSource(new MySensor());
        sensorReadingDataStreamSource.print();
        env.execute();
    }

    public static class MySensor implements SourceFunction<SensorReading> {
        private boolean running = true;

        @Override
        public void run(SourceContext<SensorReading> sourceContext) throws Exception {
            Random random = new Random();
            double gaussian = random.nextGaussian();
            HashMap<String, Double> hashMap = new HashMap<>();
            for (int i = 0; i < 10; i++) {
                hashMap.put("sensor_" + (i + 1), 60 + gaussian * 20);
            }
            while (running) {
                for (String s : hashMap.keySet()) {
                    Double aDouble = hashMap.get(s) + random.nextGaussian();
                    hashMap.put(s, aDouble);
                    sourceContext.collect(new SensorReading(s, System.currentTimeMillis(), aDouble));
                    Thread.sleep(1000L);
                }
            }
        }

        @Override
        public void cancel() {
            running = false;
        }
    }
}
